fix(proto): preserve HashJoinExec fetch across serialization - #24165

This branch is stacked on the `fetch` fix and, after the rebase onto
main, carries its own copy of the `HashJoinExec` decode path. Apply the
same correction here so the defect does not survive on this branch and
reappear once apache#24165 merges.

Review feedback on apache#24165: `u64 as usize` silently truncates on a 32-bit
target, where `usize` is 32 bits. A fetch of `1 << 32` decodes to `0` --
not merely a wrong limit but the worst possible one, since "fetch 0
rows" turns the query into an empty result instead of erroring.

Use `usize::try_from` and surface an out-of-range value as an error.
Truncating and saturating both misrepresent the plan; an explicit decode
failure is the honest outcome, and it is only reachable on a 32-bit
target with an absurd fetch.

`plan_datafusion_err!` rather than `internal_datafusion_err!`: this
decode path reserves the internal-error macros for genuinely malformed
nodes (an unknown `PartitionMode` discriminant, a dynamic filter that
does not downcast), which really do indicate a bug. A well-formed but
unrepresentable `fetch` is not a DataFusion bug -- it is a plan that
cannot be expressed on this target -- and `plan_err!` is already this
file's idiom for invalid plan configuration.

The encode side (`fetch.map(|f| f as u64)`) needs no change: `usize` is
at most 64 bits on every supported target, so widening to `u64` is
lossless.

`roundtrip_hash_join_fetch` now also covers `u32::MAX as usize` and
`usize::MAX`, pinning that a large fetch round-trips exactly. Both are
representable on every target, so the test stays portable. The
truncation path itself is only reachable on a 32-bit target and is
therefore not covered on a 64-bit CI host.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

)

## Which issue does this PR close?

- Part of apache#24171.

This addresses items (a), (b) and (c) of that issue. It does **not**
attempt (d) the colocated test tier or (e) splitting up
`roundtrip_physical_plan.rs`, both of which are larger and independent,
so the issue stays open.

Split out of apache#24167 at the reviewer's prompting: @andygrove pointed out
on that PR that the "covered by existing round-trip tests" claim did not
hold — there is no round-trip test for `SortPreservingMergeExec`. He was
right. Rather than bundle new tests with a mechanical refactor, they are
here on their own so the two can be reviewed and merged independently.

## Rationale for this change

`datafusion/proto/tests/cases/roundtrip_physical_plan.rs` is the safety
net for physical-plan serialization. A field that is silently dropped on
the wire produces a plan that still *runs* — it just returns different
rows. Checking turned up two real holes:

1. **`SortPreservingMergeExec` had no round-trip test at all.** The
string `SortPreservingMerge` did not appear anywhere in the file.
Nothing constructed one, so nothing on the encode or decode path for
that plan was exercised.

2. **`SortExec::fetch` is serialized but was never exercised.**
`roundtrip_sort` and `roundtrip_sort_preserve_partitioning` both leave
`fetch` as `None`, so the `Some(..)` state had no coverage. `fetch` is
what makes a `SortExec` a top-k sort; dropping it on the wire silently
widens the result set.

Both are cases where a serialization gap changes query results rather
than causing a visible failure.

## What changes are included in this PR?

Two new tests in `roundtrip_physical_plan.rs`. No production code
changes.

- **`roundtrip_sort_preserving_merge`** — covers everything actually on
the wire for `SortPreservingMergeExecNode`: the input, the sort
expressions, and `fetch` in both its `None` (encoded as `-1`) and
`Some(11)` states.
- **`roundtrip_sort_with_fetch`** — `SortExec` with `fetch: Some(7)`,
plus `fetch: Some(3)` combined with `preserve_partitioning: true`, since
both live in the same proto node.

### Which tests assert on accessors rather than the helper, and why

The file's `roundtrip_test` helper compares `format!("{plan:?}")` before
and after. That comparison is only as good as the plan's `Debug` impl —
it is blind to any field `Debug` does not print, which is how apache#24165
(`HashJoinExec::fetch`) survived.

`SortExec` and `SortPreservingMergeExec` both currently *derive*
`Debug`, so the helper does in fact observe `expr`, `fetch` and
`preserve_partitioning` today. I checked rather than assumed, and the
deliberate-break results below confirm it — the helper is what fires
first under each break.

But that coverage is incidental: it would vanish the day either plan
grows a hand-written `Debug`. So both new tests go through
`roundtrip_test_and_return`, downcast, and assert on `fetch()`,
`expr()`, `preserve_partitioning()` and the input schema directly, with
the helper's string comparison still running as a backstop.

### What is deliberately *not* asserted

`SortPreservingMergeExec::enable_round_robin_repartition` is **not**
serialized — `SortPreservingMergeExecNode` has only `input`, `expr` and
`fetch`, so decode always restores the `true` default from
`SortPreservingMergeExec::new`. A round-trip equality assertion would
pass whether or not that field were on the wire, so asserting on it
would advertise coverage that does not exist. The test carries a comment
saying so instead.

The same applies to `Global/LocalLimitExec::required_ordering`, which is
set by the `enforce_sorting` rule and starts as `None` on a decoded
plan.

If either field *should* be on the wire, that is a separate change with
a wire-format bump, not something to paper over with a test that cannot
tell the difference.

### Plans I checked and decided needed nothing

I went through the rest of the plans touched by apache#24167 looking for state
that is on the wire but exercised by no test. These already have
adequate coverage and I did not add to them:

| Plan | Existing coverage |
|---|---|
| `GlobalLimitExec` | `roundtrip_global_limit` (skip 0 / limit 25) and
`roundtrip_global_skip_no_limit` (skip 10 / limit `None`) — both `skip`
and `fetch` states |
| `LocalLimitExec` | `roundtrip_local_limit` |
| `FilterExec` | `roundtrip_filter_with_fetch` already asserts
`default_selectivity`, `batch_size` and `fetch` on the accessors;
`roundtrip_filter_projection_states` covers the projection |
| `ProjectionExec` | `roundtrip_projection_source`,
`roundtrip_empty_projection` |
| `RepartitionExec` | `roundtrip_repartition_preserve_order`
(round-robin + `preserve_order`), `roundtrip_range_partitioning`, plus
hash-partitioning cases |
| `UnionExec` / `InterleaveExec` | `roundtrip_union`,
`roundtrip_interleave` — nothing on the wire beyond the children |
| `CoalesceBatchesExec` | `roundtrip_coalesce_batches_with_fetch` covers
`target_batch_size` and `fetch` in both states |
| `CoalescePartitionsExec` | `roundtrip_coalesce_partitions_with_fetch`,
both `fetch` states |
| `CooperativeExec` | `roundtrip_cooperative` — only the input is on the
wire |
| `BufferExec` | `roundtrip_buffer` asserts `capacity()` on the accessor
|
| `EmptyExec` / `PlaceholderRowExec` |
`roundtrip_empty_with_partitions`,
`roundtrip_placeholder_row_with_partitions` |
| `ExplainExec` | `roundtrip_explain` asserts schema, stringified plans
and `verbose` on the accessors |
| `ScalarSubqueryExec` | `roundtrip_scalar_subquery_exec` and the
executing variant |

Per the issue, this is not a push for 100% field coverage. These two
were the cases where the gap was real and the test was cheap; past them
it got contrived fast.

## Are these changes tested?

This PR *is* tests. To confirm they are not vacuous, I broke the encode
side on purpose and checked each one fails:

| Deliberate break | Result |
|---|---|
| `SortPreservingMergeExec` encode: hardcode `fetch: -1` |
`roundtrip_sort_preserving_merge` **FAILS** |
| `SortPreservingMergeExec` encode: `.take(1)` on the sort expressions |
`roundtrip_sort_preserving_merge` **FAILS** |
| `SortExec` encode: hardcode `fetch: -1` | `roundtrip_sort_with_fetch`
**FAILS** |

Under all three breaks the pre-existing `roundtrip_sort` and
`roundtrip_sort_preserve_partitioning` kept passing — a direct
demonstration that the gap was real. All breaks reverted; the diff here
is test-only.

Checks run:

- `cargo fmt --all`
- `cargo clippy --all-targets --workspace --features
avro,integration-tests,extended_tests -- -D warnings` — clean
- `cargo test -p datafusion-proto --test proto_integration` — 216
passed, 0 failed (214 before this PR)
- `cargo test -p datafusion-physical-plan` — 1641 passed, 0 failed

### Why this is based on `main` rather than on apache#24167

These tests pass on unmodified `main` — verified, not assumed. They
describe serialization behaviour that already exists, and none of them
depends on apache#24167's changes.

That gives a useful property: merged first, they pin the current wire
behaviour independently, which makes apache#24167's "wire format unchanged"
claim something CI verifies rather than something the PR description
asserts.

## Are there any user-facing changes?

No. Test-only; no production code touched.

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

Adds the second, colocated test tier proposed in apache#24171 (d), next to the
serde hooks the plans already own.

`datafusion-physical-plan` gains `proto_test_util`, the plan-level sibling
of `datafusion_physical_expr::proto_test_util`: stub `ExecutionPlanEncode`
/ `ExecutionPlanDecode` implementations a test supplies itself, so a plan's
`try_to_proto` / `try_from_proto` can be driven without depending on
`datafusion-proto` (which sits above this crate). The stubs count their
calls and can fail on the Nth one, so the `?` arms are covered too.

On top of that, colocated `proto_tests` modules for the fields where being
wrong is expensive and `Debug`-comparing round-trip tests are blind:

* `SortExec` / `SortPreservingMergeExec` — `fetch` presence semantics
  (absent -> `None`, never `Some(0)`), `preserve_partitioning`, the
  `asc`/`descending` inversion, the TopK dynamic filter, and the reject
  paths. `SortPreservingMergeExec` also pins that
  `enable_round_robin_repartition` is deliberately not on the wire.
* `HashJoinExec` — the three `projection` states proto3 cannot express
  directly (the `[u32::MAX]` sentinel), `fetch` presence (the field dropped
  in apache#24165), the checked `u64` -> `usize` conversion, and `PartitionMode`.
* `NestedLoopJoinExec` — the same projection sentinel, which is written out
  a second time in that file and so is tested a second time here.
* `joins::proto` — exhaustive by-name round trips for `JoinType`,
  `JoinSide` and `NullEquality`, plus a guard that the two numberings
  really do differ, so nobody "simplifies" the matches into a cast.

Per apache#24171 (c) the new tests were verified against a deliberately broken
encode side rather than assumed to bite.

This supplements the central round-trip tests, it does not replace them:
those still prove the real `PhysicalExtensionCodec` works, that dispatch
reaches the hook, and that bytes survive bytes.

…24165)

## Which issue does this PR close?

<!-- No dedicated issue; found while auditing the plans covered by the
`try_to_proto`/`try_from_proto` migration EPIC. -->

- Related to apache#23494 (found while auditing that EPIC's plans for
unserialized
  fields). This is a bug fix, not part of the migration checklist.

## Rationale for this change

`HashJoinExec.fetch` was silently dropped by protobuf serialization.
`protobuf::HashJoinExecNode` had no `fetch` field, so `HashJoinExec`'s
`try_to_proto` never wrote it and `try_from_proto` never restored it: a
plan
with `fetch = Some(n)` round-tripped to `fetch = None`.

This is user-visible. The `limit_pushdown` physical optimizer rule
pushes a
limit into the join via `ExecutionPlan::with_fetch`, then marks the
global
state satisfied and drops the enclosing `GlobalLimitExec`. So after a
proto
round-trip the plan carried no limit at all, and a distributed executor
(Ballista/Comet-style, anything that ships physical plans over the wire)
returned more rows than the query asked for.

## What changes are included in this PR?

- `datafusion.proto`: add `optional uint64 fetch = 12` to
`HashJoinExecNode`.

The field is **presence-tracked on purpose**, and this is the
load-bearing
detail for wire compatibility. Messages written by versions predating
this
field carry no `fetch` at all, and a plain proto3 scalar decodes that
absence
  as `0`. With the negative-sentinel convention used by `SortExecNode`'s
`int64 fetch`, `0` would mean "fetch 0 rows" and would silently turn
every
older plan into an empty result. `optional` gives prost an `Option<u64>`
where absent decodes to `None`, which is the correct reading of an older
  message. A comment in the `.proto` records this.

- Regenerated `prost.rs` / `pbjson.rs` via
`datafusion/proto-models/regen.sh`
  (no hand edits).

- `hash_join/exec.rs`: write `self.fetch` in the `try_to_proto` hook and
restore it in `try_from_proto` via the builder's `with_fetch`, matching
how
  the plan is normally constructed.

- New regression test `roundtrip_hash_join_fetch`.

The deprecated `PhysicalPlanNodeExt` shims (`try_from_hash_join_exec` /
`try_into_hash_join_physical_plan`) delegate straight to these two
hooks, so
they pick the fix up with no separate change. Verified by reading them
rather
than assumed.

## Are these changes tested?

Yes. `roundtrip_hash_join_fetch` in
`datafusion/proto/tests/cases/roundtrip_physical_plan.rs` builds a
`HashJoinExec`, applies `with_fetch(Some(7))` the way `limit_pushdown`
does,
round-trips it through `physical_plan_to_bytes_with_proto_converter` /
`physical_plan_from_bytes_with_proto_converter`, and asserts `fetch()`
is
still `Some(7)`. It also covers `fetch = None`.

The assertion deliberately inspects `fetch()` rather than the plan's
string
form. The existing `roundtrip_test` helper compares
`format!("{plan:?}")`, and
`HashJoinExec`'s `Debug` output does not include `fetch` — which is
exactly why
this went unnoticed. I confirmed this empirically: with the encode side
reverted, the Debug comparison inside the helper still passes and only
the
`fetch()` assertion fails (`left: None, right: Some(7)`).

Ran locally:

- `cargo fmt --all`
- `cargo test -p datafusion-proto --test proto_integration` — 215
passed, 0 failed
- `cargo test -p datafusion-physical-plan` — 1640 + 9 passed, 0 failed
- `cargo clippy --all-targets --all-features` on the touched packages.
The
changed code is clean; the only two errors reported are pre-existing on
an
unmodified `main` with my newer local clippy (`uninlined_format_args` in
  `datafusion/proto-common/src/generated/pbjson.rs` and
`needless_pass_by_value` in `datafusion/proto/src/bytes/mod.rs`), in
files
  this PR does not touch.

## Are there any user-facing changes?

Yes, a bug fix: a limit pushed into a hash join now survives
physical-plan
serialization, so distributed executors no longer over-return rows. No
API
changes. The new proto field is backward and forward compatible in both
directions — old readers ignore tag 12, and new readers treat its
absence as
"no limit".

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

)

## Which issue does this PR close?

- Part of apache#24171.

This addresses items (a), (b) and (c) of that issue. It does **not**
attempt (d) the colocated test tier or (e) splitting up
`roundtrip_physical_plan.rs`, both of which are larger and independent,
so the issue stays open.

Split out of apache#24167 at the reviewer's prompting: @andygrove pointed out
on that PR that the "covered by existing round-trip tests" claim did not
hold — there is no round-trip test for `SortPreservingMergeExec`. He was
right. Rather than bundle new tests with a mechanical refactor, they are
here on their own so the two can be reviewed and merged independently.

## Rationale for this change

`datafusion/proto/tests/cases/roundtrip_physical_plan.rs` is the safety
net for physical-plan serialization. A field that is silently dropped on
the wire produces a plan that still *runs* — it just returns different
rows. Checking turned up two real holes:

1. **`SortPreservingMergeExec` had no round-trip test at all.** The
string `SortPreservingMerge` did not appear anywhere in the file.
Nothing constructed one, so nothing on the encode or decode path for
that plan was exercised.

2. **`SortExec::fetch` is serialized but was never exercised.**
`roundtrip_sort` and `roundtrip_sort_preserve_partitioning` both leave
`fetch` as `None`, so the `Some(..)` state had no coverage. `fetch` is
what makes a `SortExec` a top-k sort; dropping it on the wire silently
widens the result set.

Both are cases where a serialization gap changes query results rather
than causing a visible failure.

## What changes are included in this PR?

Two new tests in `roundtrip_physical_plan.rs`. No production code
changes.

- **`roundtrip_sort_preserving_merge`** — covers everything actually on
the wire for `SortPreservingMergeExecNode`: the input, the sort
expressions, and `fetch` in both its `None` (encoded as `-1`) and
`Some(11)` states.
- **`roundtrip_sort_with_fetch`** — `SortExec` with `fetch: Some(7)`,
plus `fetch: Some(3)` combined with `preserve_partitioning: true`, since
both live in the same proto node.

### Which tests assert on accessors rather than the helper, and why

The file's `roundtrip_test` helper compares `format!("{plan:?}")` before
and after. That comparison is only as good as the plan's `Debug` impl —
it is blind to any field `Debug` does not print, which is how apache#24165
(`HashJoinExec::fetch`) survived.

`SortExec` and `SortPreservingMergeExec` both currently *derive*
`Debug`, so the helper does in fact observe `expr`, `fetch` and
`preserve_partitioning` today. I checked rather than assumed, and the
deliberate-break results below confirm it — the helper is what fires
first under each break.

But that coverage is incidental: it would vanish the day either plan
grows a hand-written `Debug`. So both new tests go through
`roundtrip_test_and_return`, downcast, and assert on `fetch()`,
`expr()`, `preserve_partitioning()` and the input schema directly, with
the helper's string comparison still running as a backstop.

### What is deliberately *not* asserted

`SortPreservingMergeExec::enable_round_robin_repartition` is **not**
serialized — `SortPreservingMergeExecNode` has only `input`, `expr` and
`fetch`, so decode always restores the `true` default from
`SortPreservingMergeExec::new`. A round-trip equality assertion would
pass whether or not that field were on the wire, so asserting on it
would advertise coverage that does not exist. The test carries a comment
saying so instead.

The same applies to `Global/LocalLimitExec::required_ordering`, which is
set by the `enforce_sorting` rule and starts as `None` on a decoded
plan.

If either field *should* be on the wire, that is a separate change with
a wire-format bump, not something to paper over with a test that cannot
tell the difference.

### Plans I checked and decided needed nothing

I went through the rest of the plans touched by apache#24167 looking for state
that is on the wire but exercised by no test. These already have
adequate coverage and I did not add to them:

| Plan | Existing coverage |
|---|---|
| `GlobalLimitExec` | `roundtrip_global_limit` (skip 0 / limit 25) and
`roundtrip_global_skip_no_limit` (skip 10 / limit `None`) — both `skip`
and `fetch` states |
| `LocalLimitExec` | `roundtrip_local_limit` |
| `FilterExec` | `roundtrip_filter_with_fetch` already asserts
`default_selectivity`, `batch_size` and `fetch` on the accessors;
`roundtrip_filter_projection_states` covers the projection |
| `ProjectionExec` | `roundtrip_projection_source`,
`roundtrip_empty_projection` |
| `RepartitionExec` | `roundtrip_repartition_preserve_order`
(round-robin + `preserve_order`), `roundtrip_range_partitioning`, plus
hash-partitioning cases |
| `UnionExec` / `InterleaveExec` | `roundtrip_union`,
`roundtrip_interleave` — nothing on the wire beyond the children |
| `CoalesceBatchesExec` | `roundtrip_coalesce_batches_with_fetch` covers
`target_batch_size` and `fetch` in both states |
| `CoalescePartitionsExec` | `roundtrip_coalesce_partitions_with_fetch`,
both `fetch` states |
| `CooperativeExec` | `roundtrip_cooperative` — only the input is on the
wire |
| `BufferExec` | `roundtrip_buffer` asserts `capacity()` on the accessor
|
| `EmptyExec` / `PlaceholderRowExec` |
`roundtrip_empty_with_partitions`,
`roundtrip_placeholder_row_with_partitions` |
| `ExplainExec` | `roundtrip_explain` asserts schema, stringified plans
and `verbose` on the accessors |
| `ScalarSubqueryExec` | `roundtrip_scalar_subquery_exec` and the
executing variant |

Per the issue, this is not a push for 100% field coverage. These two
were the cases where the gap was real and the test was cheap; past them
it got contrived fast.

## Are these changes tested?

This PR *is* tests. To confirm they are not vacuous, I broke the encode
side on purpose and checked each one fails:

| Deliberate break | Result |
|---|---|
| `SortPreservingMergeExec` encode: hardcode `fetch: -1` |
`roundtrip_sort_preserving_merge` **FAILS** |
| `SortPreservingMergeExec` encode: `.take(1)` on the sort expressions |
`roundtrip_sort_preserving_merge` **FAILS** |
| `SortExec` encode: hardcode `fetch: -1` | `roundtrip_sort_with_fetch`
**FAILS** |

Under all three breaks the pre-existing `roundtrip_sort` and
`roundtrip_sort_preserve_partitioning` kept passing — a direct
demonstration that the gap was real. All breaks reverted; the diff here
is test-only.

Checks run:

- `cargo fmt --all`
- `cargo clippy --all-targets --workspace --features
avro,integration-tests,extended_tests -- -D warnings` — clean
- `cargo test -p datafusion-proto --test proto_integration` — 216
passed, 0 failed (214 before this PR)
- `cargo test -p datafusion-physical-plan` — 1641 passed, 0 failed

### Why this is based on `main` rather than on apache#24167

These tests pass on unmodified `main` — verified, not assumed. They
describe serialization behaviour that already exists, and none of them
depends on apache#24167's changes.

That gives a useful property: merged first, they pin the current wire
behaviour independently, which makes apache#24167's "wire format unchanged"
claim something CI verifies rather than something the PR description
asserts.

## Are there any user-facing changes?

No. Test-only; no production code touched.

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

…he#24222)

Adds the second, colocated test tier proposed in apache#24171 (d), next to the
serde hooks the plans already own.

`datafusion-physical-plan` gains `proto_test_util`, the plan-level
sibling of `datafusion_physical_expr::proto_test_util`: stub
`ExecutionPlanEncode` / `ExecutionPlanDecode` implementations a test
supplies itself, so a plan's `try_to_proto` / `try_from_proto` can be
driven without depending on `datafusion-proto` (which sits above this
crate). The stubs count their calls and can fail on the Nth one, so the
`?` arms are covered too.

On top of that, colocated `proto_tests` modules for the fields where
being wrong is expensive and `Debug`-comparing round-trip tests are
blind:

* `SortExec` / `SortPreservingMergeExec` — `fetch` presence semantics
(absent -> `None`, never `Some(0)`), `preserve_partitioning`, the
`asc`/`descending` inversion, the TopK dynamic filter, and the reject
paths. `SortPreservingMergeExec` also pins that
`enable_round_robin_repartition` is deliberately not on the wire.
* `HashJoinExec` — the three `projection` states proto3 cannot express
directly (the `[u32::MAX]` sentinel), `fetch` presence (the field
dropped in apache#24165), the checked `u64` -> `usize` conversion, and
`PartitionMode`.
* `NestedLoopJoinExec` — the same projection sentinel, which is written
out a second time in that file and so is tested a second time here.
* `joins::proto` — exhaustive by-name round trips for `JoinType`,
`JoinSide` and `NullEquality`, plus a guard that the two numberings
really do differ, so nobody "simplifies" the matches into a cast.

Per apache#24171 (c) the new tests were verified against a deliberately broken
encode side rather than assumed to bite.

This supplements the central round-trip tests, it does not replace them:
those still prove the real `PhysicalExtensionCodec` works, that dispatch
reaches the hook, and that bytes survive bytes.

---------

Co-authored-by: Claude <noreply@anthropic.com>

…ks (apache#24164)

## Which issue does this PR close?

<!-- No dedicated issue; this is follow-up cleanup for the proto hook
migration EPIC. -->

- Part of apache#23494.

## Rationale for this change

EPIC apache#23494 moved every built-in `ExecutionPlan` off the central
`downcast_ref` chain
in `datafusion-proto` onto per-plan hooks: a `try_to_proto` override
inside
`impl ExecutionPlan for FooExec`, and an inherent
`FooExec::try_from_proto`.

Those hooks read plan state through **getters** (`self.filter()`,
`self.join_type()`, …).
That makes adding a field to a plan struct invisible to serialization —
nothing breaks
at compile time, the new field is just silently not serialized, and the
bug only shows
up as state quietly disappearing on a round-trip.

This is not hypothetical. `HashJoinExec.fetch` was dropped on round-trip
for exactly
this reason (fixed in apache#24165) — the field was added to the struct and
the encoder,
still calling getters, never learned about it.

Destructuring removes the failure mode. If `try_to_proto` starts with an
exhaustive
`let Self { … } = self;` with **no `..`**, adding a field to the plan
struct is a compile
error in the encoder. If `try_from_proto` destructures the
prost-generated node struct the
same way, adding a field to the `.proto` is a compile error in every
decoder. The prost
structs are plain, all-`pub`, and not `#[non_exhaustive]`, so this
compiles today.

## What changes are included in this PR?

Applies the pattern to the join plans, one commit per plan:

- `CrossJoinExec`
- `NestedLoopJoinExec`
- `SortMergeJoinExec`
- `SymmetricHashJoinExec`
- `HashExpr` / `HashTableLookupExpr`
(`joins/hash_join/partitioned_hash_eval.rs`)
- `HashJoinExec`

Fields that genuinely are not serialized bind to `_` and carry a short
comment saying
why — derived at construction, runtime state, or recomputed by
`new`/`try_new`/the builder
on decode. For example:

```rust
let Self {
    left,
    right,
    on,
    filter,
    join_type,
    sort_options,
    null_equality,
    // derived from the children's schemas by `try_new` on decode
    schema: _,
    // runtime metrics, not part of the plan
    metrics: _,
    // recomputed from `on` and `sort_options` by `try_new` on decode
    left_sort_exprs: _,
    // recomputed from `on` and `sort_options` by `try_new` on decode
    right_sort_exprs: _,
    // recomputed by `try_new` on decode
    cache: _,
} = self;
```

`HashTableLookupExpr::try_to_proto` deliberately serializes none of its
state — it holds a
runtime `Arc<Map>` and is replaced with `lit(true)`. It gets a
destructure with every field
bound to `_`, so that adding a field there forces a decision instead of
passing unnoticed.

For `HashJoinExec`, the serialized bindings are `left`, `right`, `on`,
`filter`,
`join_type`, `mode`, `projection`, `null_equality`, `null_aware`,
`dynamic_filter` and
`fetch`. The six ignored fields are `join_schema`, `left_fut`,
`random_state`, `metrics`,
`column_indices` and `cache` — `random_state` is the fixed
`HASH_JOIN_SEED` constant that
the builder sets identically on decode, and the rest are runtime state
or recomputed by the
builder.

**No additional unserialized fields were found.** Every field on the
join plans that is
not written to the proto is genuinely derived, recomputed on decode, or
runtime-only.

**This is a pure refactor. The wire format is byte-for-byte unchanged
and no behavior changes.**
Every getter that was replaced returns exactly the field it was replaced
with. The only
incidental changes are in `SymmetricHashJoinExec::try_from_proto` and
`HashJoinExec::try_from_proto`, where four `internal_datafusion_err!`
calls switched to
inlined format args now that the field is bound locally; the message
text they produce is
identical.

## Are these changes tested?

Covered by the existing round-trip tests — that is the point of the
change: if the
destructures had drifted from the encoders, the plans would fail to
round-trip.

Run locally:

- `cargo test -p datafusion-proto --test proto_integration` — 215
passed, including all
five hash-join round-trip tests (`roundtrip_hash_join`,
`roundtrip_hash_join_fetch`,
`roundtrip_hash_join_projection_states`,
`test_hash_join_with_dynamic_filter_roundtrip`,
  `roundtrip_sym_hash_join`)
- `cargo test -p datafusion-physical-plan` — 1641 passed
- `cargo clippy --all-targets --workspace --features
avro,integration-tests,extended_tests -- -D warnings`
- `cargo fmt --all`

## Are there any user-facing changes?

No.

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

…e hooks (apache#24167)

## Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes apache#123` indicates that this PR will close issue apache#123.
-->

- Follow-up cleanup for EPIC apache#23494; does not close an issue on its own.

## Rationale for this change

EPIC apache#23494 moved every built-in `ExecutionPlan` off the central
`downcast_ref`
chain in `datafusion-proto` onto per-plan hooks: a `try_to_proto`
override on
the `ExecutionPlan` trait and an inherent `FooExec::try_from_proto`,
both living
in the plan's own module.

Those hooks currently read plan state through **getters**
(`self.expr()`,
`self.fetch()`, ...). That means adding a field to a plan struct is
invisible to
serialization: nothing breaks, nothing warns, the field is just silently
dropped on round-trip. This is not hypothetical — it has already cost us
a real
bug: `HashJoinExec.fetch` is not serialized and is lost on round-trip
(being
fixed separately).

Starting each hook with an *exhaustive* destructure turns that class of
mistake
into a compile error:

* **Encode side** — destructure `self` naming every field, no `..`. Add
a field
to `SortExec` and `SortExec::try_to_proto` stops compiling until you
either
serialize it or bind it to `_` with a comment saying why it does not
need to
  be (derived at construction, runtime state, recomputed on decode).
* **Decode side** — destructure the prost node struct the same way. The
generated structs are plain, all-`pub` and not `#[non_exhaustive]`, so
adding
a field to the `.proto` becomes a compile error in every decoder rather
than a
  silent omission.

## What changes are included in this PR?

A mechanical, behavior-preserving refactor of the serde hooks for the
core
(non-join, non-aggregate, non-window) plans in
`datafusion/physical-plan/src/`:

`SortExec`, `SortPreservingMergeExec`, `GlobalLimitExec`,
`LocalLimitExec`,
`FilterExec`, `ProjectionExec`, `RepartitionExec`, `UnionExec`,
`InterleaveExec`, `CoalesceBatchesExec`, `CoalescePartitionsExec`,
`CooperativeExec`, `BufferExec`, `EmptyExec`, `PlaceholderRowExec`,
`ExplainExec`, `ScalarSubqueryExec`.

**The wire format is unchanged — byte for byte — and no behavior
changes.** The
same values are written to and read from the same proto fields; only the
way
they are reached in Rust changes.

Split into four commits, one per group of plans, each of which builds
green.

### Fields the refactor documented as not serialized

Writing the destructures surfaced and pinned down, in comments at the
point of
use, which fields do not survive a round-trip today. All of these are
pre-existing; **none of them are changed here**, because fixing any of
them
means changing the wire format, which is a separate decision:

* `SortPreservingMergeExec::enable_round_robin_repartition` — no field
on
`SortPreservingMergeExecNode`; decoding always restores the `true`
default
  from `SortPreservingMergeExec::new`.
* `GlobalLimitExec::required_ordering` and
`LocalLimitExec::required_ordering` —
no field on the limit nodes; they are set by the `enforce_sorting`
optimizer
  rule, so a decoded plan starts with `None`.

Everything else that binds to `_` is genuinely not plan shape: runtime
metrics
(`metrics` / `metrics_set`), execution-time state
(`RepartitionExec::state`,
`ScalarSubqueryExec::subquery_future` and `results`), values derived at
construction from other serialized fields
(`SortExec::common_sort_prefix`), the
positional `ScalarSubqueryLink::index`, and the `cache` / `properties`
plan
properties that are recomputed on decode.

Two small notes for reviewers:

* `RepartitionExec` keeps its output partitioning *inside* `cache`
(`partitioning()` is `&self.cache.partitioning`), so `cache` is bound
rather
  than `_`-ed there and is read for the serialized `partitioning` field.
* `EmptyExec` / `PlaceholderRowExec` now read the encoded partition
count from
  the `partitions` field instead of going through
`properties().output_partitioning().partition_count()`. The two are kept
in
sync by `with_partitions`, the only writer besides `new`, so the encoded
value
  is identical.

## Are these changes tested?

Covered by the existing round-trip tests, which are the real proof that
the wire
format did not move. Locally on this branch:

* `cargo test -p datafusion-proto --test proto_integration` — 214
passed, 0 failed
* `cargo test -p datafusion-physical-plan` — 1640 passed, 0 failed
* `cargo clippy -p datafusion-physical-plan --all-targets --all-features
-- -D warnings` — clean
* `cargo fmt --all`

No new tests are added here: the change adds no new behavior to test,
and the
compile-error property it introduces is enforced by rustc rather than by
a test.

**On the comprehensiveness of that existing coverage** (thanks
@andygrove for
pushing on this — the original wording oversold it):

* `SortPreservingMergeExec`, updated by this PR, has **no** round-trip
test at
  all — `SortPreservingMerge` does not appear anywhere in
`roundtrip_physical_plan.rs`. The other 16 plans touched here each have
at
  least one dedicated `roundtrip_*` test.
* Separately, `roundtrip_test` asserts on `format!("{plan:?}")`, so it
cannot
catch a dropped field that a plan's `Debug` impl doesn't print. That is
  precisely how the `HashJoinExec::fetch` bug (apache#24165) went unnoticed.

Both are now tracked in apache#24171, with the test work split out into its
own PR so
this mechanical refactor and the coverage improvements can be reviewed
independently. Sequencing is up to reviewers — happy to land the tests
first if
that's preferred.

## Are there any user-facing changes?

No. No public API changes, no wire format changes, no behavior changes.

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>